package com.tyjt.common.mqtt.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.Executors;

/**
 *
 */
@Slf4j
@Data
@Configuration
public class MqttConfig {

    @Value("${mqtt.client.username}")
    private String username;

    @Value("${mqtt.client.password}")
    private String password;

    @Value("${mqtt.client.serverURI}")
    private String serverURI;

    @Value("${mqtt.client.clientId}")
    private String clientId;

    @Value("${mqtt.client.group:g1}")
    private String groupId;

    @Bean
    public MqttClient mqttClient(MqttClientPersistence persistence) throws MqttException {

        try {
            MqttClient mqttClient = new MqttClient(serverURI, clientId, persistence);

            // QoS0，At most once，至多一次；
            // QoS1，At least once，至少一次；
            // QoS2，Exactly once，确保只有一次。
            MqttConnectOptions connectOptions = getMqttConnectOptions();

            MqttCallBack mqttCallBack = new MqttCallBack(mqttClient);
            mqttClient.setCallback(mqttCallBack);

            connectOptions.setKeepAliveInterval(60);
            mqttClient.connect(connectOptions);

            return mqttClient;

        } catch (MqttException mqttException) {
            log.error("初始化mqtt client失败！ serverURI  {} ", serverURI, mqttException);
            throw mqttException;
        }

    }

    private MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        connectOptions.setUserName(username);
        connectOptions.setPassword(password.toCharArray());
        connectOptions.setCleanSession(true);
        connectOptions.setAutomaticReconnect(true);
        return connectOptions;
    }

    @Bean
    @ConditionalOnMissingBean(MqttClientPersistence.class)
    public MqttClientPersistence mqttClientPersistence() {
        return new MemoryPersistence();
    }

    @Bean
    public MqttListenerAnnotationBeanPostProcessor mqttListenerAnnotationBeanPostProcessor() {
        return new MqttListenerAnnotationBeanPostProcessor(groupId);
    }

}
